<!DOCTYPE HTML>
<!--
	Dimension by HTML5 UP
	html5up.net | @ajlkn
	Free for personal and commercial use under the CCA 3.0 license (html5up.net/license)
-->
<html>
 <head>
  <title>
   Dimension by HTML5 UP
  </title>
  <!-- <meta charset="utf-8" /> -->
  <!-- <meta name="viewport" content="width=device-width, initial-scale=1, user-scalable=no" /> -->
  <meta charset="utf-8"/>
  <meta content="width=device-width,initial-scale=1.0" name="viewport"/>
  <link href="../../assets/css/article.css" rel="stylesheet"/>
  <link href="https://cdn.bootcss.com/highlight.js/9.15.8/styles/github.min.css" rel="stylesheet"/>
  <noscript>
   <link href="../../assets/css/noscript.css" rel="stylesheet"/>
  </noscript>
 </head>
 <body>
  <div id="app">
  </div>
  <!-- built files will be auto injected -->
 </body>
 <body class="is-preload">
  <!-- Wrapper -->
  <div id="wrapper">
   <!-- Main -->
   <div id="main">
    <article id="article">
     <h1 id="kafka-sasl">
      Kafka SASL 安全验证
     </h1>
     <hr/>
     <p>
      这是我参与2022首次更文挑战的第11天，活动详情查看：
      <a href="https://juejin.cn/post/7052884569032392740">
       2022首次更文挑战
      </a>
     </p>
     <p>
      <em>
       操作都在相应的软件安装目录下
      </em>
     </p>
     <h2 id="_1">
      配置文件编写
     </h2>
     <h3 id="kafka-server-jaasconf">
      服务配置文件kafka-server-jaas.conf
     </h3>
     <div class="codehilite">
      <pre><span></span><code>vim config/kafka-server-jaas.conf

<span class="c1"># add</span>
KafkaServer <span class="o">{</span>
    org.apache.kafka.common.security.plain.PlainLoginModule required
    <span class="nv">username</span><span class="o">=</span><span class="s2">"admin"</span>
    <span class="nv">password</span><span class="o">=</span><span class="s2">"admin"</span>
    <span class="nv">user_admin</span><span class="o">=</span><span class="s2">"admin"</span>
    <span class="nv">user_alice</span><span class="o">=</span><span class="s2">"alice"</span><span class="p">;</span>
<span class="o">}</span><span class="p">;</span>

KafkaClient <span class="o">{</span>
  org.apache.kafka.common.security.plain.PlainLoginModule required
  <span class="nv">username</span><span class="o">=</span><span class="s2">"alice"</span>
  <span class="nv">password</span><span class="o">=</span><span class="s2">"alice"</span><span class="p">;</span>
<span class="o">}</span><span class="p">;</span>
</code></pre>
     </div>
     <h3 id="serverproperties">
      启动配置文件server.properties
     </h3>
     <div class="codehilite">
      <pre><span></span><code>cp config/server.properties config/sasl-server.properties

vim config/sasl-server.properties

<span class="c1"># 在末尾添加</span>
<span class="c1"># 遇到外网端口不放连接的情况,需要直接判断外网ip和对应的开发的外网端口,好像topic自动创建会失效</span>
advertised.listeners<span class="o">=</span>SASL_PLAINTEXT://10.0.0.51:9092
<span class="nv">listeners</span><span class="o">=</span>SASL_PLAINTEXT://10.0.0.51:9092
security.inter.broker.protocol<span class="o">=</span>SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol<span class="o">=</span>PLAIN
sasl.enabled.mechanisms<span class="o">=</span>PLAIN
</code></pre>
     </div>
     <h3 id="kafka-server-startsh">
      启动文件kafka-server-start.sh
     </h3>
     <div class="codehilite">
      <pre><span></span><code>cp bin/kafka-server-start.sh bin/sasl-kafka-server-start.sh
vim bin/sasl-kafka-server-start.sh

<span class="c1"># change the line from</span>
<span class="k">if</span> <span class="o">[</span> <span class="s2">"x</span><span class="nv">$KAFKA_HEAP_OPTS</span><span class="s2">"</span> <span class="o">=</span> <span class="s2">"x"</span> <span class="o">]</span><span class="p">;</span> <span class="k">then</span>
    <span class="nb">export</span> <span class="nv">KAFKA_HEAP_OPTS</span><span class="o">=</span><span class="s2">"-Xmx1G -Xms1G"</span>
<span class="k">fi</span>

<span class="c1"># to</span>
<span class="k">if</span> <span class="o">[</span> <span class="s2">"x</span><span class="nv">$KAFKA_HEAP_OPTS</span><span class="s2">"</span> <span class="o">=</span> <span class="s2">"x"</span> <span class="o">]</span><span class="p">;</span> <span class="k">then</span>
    <span class="nb">export</span> <span class="nv">KAFKA_HEAP_OPTS</span><span class="o">=</span><span class="s2">"-Xmx1G -Xms1G -Djava.security.auth.login.config=</span><span class="nv">$base_dir</span><span class="s2">/../config/kafka-server-jaas.conf"</span>
<span class="k">fi</span>
</code></pre>
     </div>
     <h3 id="_2">
      生产者相关文件修改
     </h3>
     <h4 id="_3">
      新建生产者客户端配置文件
     </h4>
     <div class="codehilite">
      <pre><span></span><code>vim ./config/jaas-kafka-client.conf

KafkaClient <span class="o">{</span>
  org.apache.kafka.common.security.plain.PlainLoginModule required
  <span class="nv">username</span><span class="o">=</span><span class="s2">"alice"</span>
  <span class="nv">password</span><span class="o">=</span><span class="s2">"alice"</span><span class="p">;</span>
<span class="o">}</span><span class="p">;</span>
</code></pre>
     </div>
     <h3 id="_4">
      生产者配置文件修改
     </h3>
     <div class="codehilite">
      <pre><span></span><code>cp ./config/producer.properties ./config/sasl-producer.properties
vim ./config/sasl-producer.properties

<span class="c1">#修改bootstrap.servers</span>
bootstrap.servers<span class="o">=</span><span class="m">10</span>.0.0.51:9092

<span class="c1"># 在末尾添加</span>
security.protocol<span class="o">=</span>SASL_PLAINTEXT
sasl.mechanism<span class="o">=</span>PLAIN
</code></pre>
     </div>
     <h4 id="_5">
      生产者启动脚本修改
     </h4>
     <div class="codehilite">
      <pre><span></span><code>cp ./bin/kafka-console-producer.sh ./bin/sasl-kafka-console-producer.sh
vim ./bin/sasl-kafka-console-producer.sh

<span class="c1"># modifiy below</span>
 <span class="k">if</span> <span class="o">[</span> <span class="s2">"x</span><span class="nv">$KAFKA_HEAP_OPTS</span><span class="s2">"</span> <span class="o">=</span> <span class="s2">"x"</span> <span class="o">]</span><span class="p">;</span> <span class="k">then</span>
    <span class="nb">export</span> <span class="nv">KAFKA_HEAP_OPTS</span><span class="o">=</span><span class="s2">"-Xmx512M"</span>
 <span class="k">fi</span>
 <span class="nb">exec</span> <span class="k">$(</span>dirname <span class="nv">$0</span><span class="k">)</span>/kafka-run-class.sh kafka.tools.ConsoleProducer <span class="s2">"</span><span class="nv">$@</span><span class="s2">"</span>

<span class="c1"># to</span>
 <span class="k">if</span> <span class="o">[</span> <span class="s2">"x</span><span class="nv">$KAFKA_HEAP_OPTS</span><span class="s2">"</span> <span class="o">=</span> <span class="s2">"x"</span> <span class="o">]</span><span class="p">;</span> <span class="k">then</span>
    <span class="nb">export</span> <span class="nv">KAFKA_HEAP_OPTS</span><span class="o">=</span><span class="s2">"-Xmx512M"</span>
 <span class="k">fi</span>
 <span class="nb">exec</span> <span class="k">$(</span>dirname <span class="nv">$0</span><span class="k">)</span>/kafka-run-class.sh -Djava.security.auth.login.config<span class="o">=</span><span class="k">$(</span>dirname <span class="nv">$0</span><span class="k">)</span>/../config/jaas-kafka-client.conf kafka.tools.ConsoleProducer <span class="s2">"</span><span class="nv">$@</span><span class="s2">"</span>
</code></pre>
     </div>
     <h3 id="_6">
      消费者相关修改
     </h3>
     <h4 id="_7">
      新建生产者客户端配置文件
     </h4>
     <div class="codehilite">
      <pre><span></span><code>vim ./config/jaas-kafka-consumer.conf

KafkaServer <span class="o">{</span>
    org.apache.kafka.common.security.plain.PlainLoginModule required
    <span class="nv">username</span><span class="o">=</span><span class="s2">"admin"</span>
    <span class="nv">password</span><span class="o">=</span><span class="s2">"admin"</span>
    <span class="nv">user_admin</span><span class="o">=</span><span class="s2">"admin"</span>
    <span class="nv">user_alice</span><span class="o">=</span><span class="s2">"alice"</span><span class="p">;</span>
<span class="o">}</span><span class="p">;</span>

KafkaClient <span class="o">{</span>
  org.apache.kafka.common.security.plain.PlainLoginModule required
  <span class="nv">username</span><span class="o">=</span><span class="s2">"alice"</span>
  <span class="nv">password</span><span class="o">=</span><span class="s2">"alice"</span><span class="p">;</span>
<span class="o">}</span><span class="p">;</span>
</code></pre>
     </div>
     <h4 id="_8">
      新建消费者配置文件
     </h4>
     <div class="codehilite">
      <pre><span></span><code>cp ./config/consumer.properties ./config/sasl-consumer.properties
vim ./config/sasl-consumer.properties

<span class="c1">#修改bootstrap.servers</span>
bootstrap.servers<span class="o">=</span><span class="m">10</span>.0.0.51:9092

<span class="c1"># 末尾添加</span>
security.protocol<span class="o">=</span>SASL_PLAINTEXT
sasl.mechanism<span class="o">=</span>PLAIN
</code></pre>
     </div>
     <h4 id="_9">
      消费者启动脚本修改
     </h4>
     <div class="codehilite">
      <pre><span></span><code>cp ./bin/kafka-console-consumer.sh ./bin/sasl-kafka-console-consumer.sh
vim ./bin/sasl-kafka-console-consumer.sh

<span class="c1"># modifiy below</span>
<span class="k">if</span> <span class="o">[</span> <span class="s2">"x</span><span class="nv">$KAFKA_HEAP_OPTS</span><span class="s2">"</span> <span class="o">=</span> <span class="s2">"x"</span> <span class="o">]</span><span class="p">;</span> <span class="k">then</span>
   <span class="nb">export</span> <span class="nv">KAFKA_HEAP_OPTS</span><span class="o">=</span><span class="s2">"-Xmx512M"</span>
<span class="k">fi</span>
<span class="nb">exec</span> <span class="k">$(</span>dirname <span class="nv">$0</span><span class="k">)</span>/kafka-run-class.sh kafka.tools.ConsoleConsumer <span class="s2">"</span><span class="nv">$@</span><span class="s2">"</span>

<span class="c1"># to</span>
<span class="k">if</span> <span class="o">[</span> <span class="s2">"x</span><span class="nv">$KAFKA_HEAP_OPTS</span><span class="s2">"</span> <span class="o">=</span> <span class="s2">"x"</span> <span class="o">]</span><span class="p">;</span> <span class="k">then</span>
   <span class="nb">export</span> <span class="nv">KAFKA_HEAP_OPTS</span><span class="o">=</span><span class="s2">"-Xmx512M"</span>
<span class="k">fi</span>
<span class="nb">exec</span> <span class="k">$(</span>dirname <span class="nv">$0</span><span class="k">)</span>/kafka-run-class.sh -Djava.security.auth.login.config<span class="o">=</span><span class="k">$(</span>dirname <span class="nv">$0</span><span class="k">)</span>/../config/jaas-kafka-consumer.conf kafka.tools.ConsoleConsumer <span class="s2">"</span><span class="nv">$@</span><span class="s2">"</span>
 <span class="sb">```</span>

<span class="c1">## 集群配置</span>
可以复制整个zookeeper和kafka的文件夹,修改配置文件的中kafka配置文件的broken.id和zookeeper的myid文件内容

<span class="c1">## 安全验证后使用的生产和消费命令</span>
- /usr/local/kafka/bin/sasl-kafka-server-start.sh -daemon /usr/local/kafka/config/sasl-server.properties

- /usr/local/kafka/bin/sasl-kafka-console-producer.sh --broker-list <span class="m">172</span>.19.104.18:9092,172.19.104.19:9092,172.19.104.21:9092 --topic <span class="nb">test</span> --producer.config /usr/local/kafka/config/sasl-producer.properties

- /usr/local/kafka/bin/sasl-kafka-console-consumer.sh --bootstrap-server <span class="m">172</span>.19.104.18:9092,172.19.104.19:9092,172.19.104.21:9092 --topic <span class="nb">test</span> --consumer.config /usr/local/kafka/config/sasl-consumer.properties

- /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper <span class="m">172</span>.19.104.18:2181,172.19.104.19:2181,172.19.104.21:2181

- /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper <span class="m">172</span>.19.104.18:2181,172.19.104.19:2181,172.19.104.21:2181 --topic test1 --replication-factor <span class="m">2</span> --partitions <span class="m">3</span>

<span class="c1">## 编程调用</span>
<span class="c1">### Python</span>
<span class="sb">```</span>python
<span class="c1"># 消费者</span>
<span class="nv">consumer</span> <span class="o">=</span> KafkaConsumer<span class="o">(</span><span class="nv">bootstrap_servers</span><span class="o">=</span><span class="s1">'172.19.104.18:9092,172.19.104.19:9092,172.19.104.21:9092'</span>, <span class="nv">security_protocol</span><span class="o">=</span><span class="s2">"SASL_PLAINTEXT"</span>, <span class="nv">sasl_mechanism</span><span class="o">=</span><span class="s1">'PLAIN'</span>, <span class="nv">sasl_plain_username</span><span class="o">=</span><span class="s1">'alice'</span>, <span class="nv">sasl_plain_password</span><span class="o">=</span><span class="s1">'alice'</span><span class="o">)</span>
consumer.subscribe<span class="o">([</span><span class="s2">"test"</span><span class="o">])</span>
<span class="k">for</span> msg in consumer:
    print<span class="o">(</span>str<span class="o">(</span>msg<span class="o">[</span><span class="m">6</span><span class="o">]</span>, <span class="nv">encoding</span><span class="o">=</span><span class="s2">"utf-8"</span><span class="o">))</span>

<span class="c1"># 生产者</span>
<span class="nv">producer</span> <span class="o">=</span> KafkaProducer<span class="o">(</span><span class="nv">bootstrap_servers</span><span class="o">=</span><span class="s1">'172.19.104.18:9092,172.19.104.19:9092,172.19.104.21:9092'</span>, <span class="nv">security_protocol</span><span class="o">=</span><span class="s2">"SASL_PLAINTEXT"</span>, <span class="nv">sasl_mechanism</span><span class="o">=</span><span class="s1">'PLAIN'</span>, <span class="nv">sasl_plain_username</span><span class="o">=</span><span class="s1">'alice'</span>, <span class="nv">sasl_plain_password</span><span class="o">=</span><span class="s1">'alice'</span><span class="o">)</span>
<span class="nv">msg</span> <span class="o">=</span> <span class="o">{</span><span class="s2">"test"</span>: <span class="s2">"test"</span><span class="o">}</span>
producer.send<span class="o">(</span><span class="s2">"test"</span>, bytes<span class="o">(</span>str<span class="o">(</span>msg<span class="o">)</span>.replace<span class="o">(</span><span class="s2">"'"</span>, <span class="s1">'"'</span><span class="o">)</span>, <span class="s2">"utf-8"</span><span class="o">))</span>
time.sleep<span class="o">(</span><span class="m">1</span><span class="o">)</span>
</code></pre>
     </div>
     <h3 id="java">
      Java
     </h3>
     <p>
      <em>
       kafka_server_jaas.conf kafka_client_jaas.conf是上面新建文件：jaas-kafka-client.conf jaas-kafka-consumer.conf的复制，对应的原理是将kafka验证文件进行加载读取即可
      </em>
     </p>
     <div class="codehilite">
      <pre><span></span><code>    <span class="cm">/***</span>
<span class="cm">     * 生成特定Topic的消费者并返回(为了避免消息丢失，采用手动提交位移方式）</span>
<span class="cm">     * @param topic</span>
<span class="cm">     * @param groupName</span>
<span class="cm">     * @return</span>
<span class="cm">     */</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="n">KafkaConsumer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="nf">createConsumer</span><span class="p">(</span><span class="n">String</span> <span class="n">topic</span><span class="p">,</span> <span class="n">String</span> <span class="n">groupName</span><span class="p">)</span> <span class="p">{</span>
        <span class="n">logger</span><span class="p">.</span><span class="na">info</span><span class="p">(</span><span class="s">"Create kafka consumer"</span><span class="p">);</span>
        <span class="n">Properties</span> <span class="n">properties</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="p">();</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"bootstrap.servers"</span><span class="p">,</span> <span class="n">Holder</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="s">"kafka.host"</span><span class="p">));</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"group.id"</span><span class="p">,</span> <span class="n">groupName</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"enable.auto.commit"</span><span class="p">,</span> <span class="s">"true"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"auto.commit.interval.ms"</span><span class="p">,</span> <span class="s">"1000"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"auto.offset.reset"</span><span class="p">,</span> <span class="s">"latest"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"session.timeout.ms"</span><span class="p">,</span> <span class="s">"30000"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"key.deserializer"</span><span class="p">,</span> <span class="s">"org.apache.kafka.common.serialization.StringDeserializer"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"value.deserializer"</span><span class="p">,</span> <span class="s">"org.apache.kafka.common.serialization.StringDeserializer"</span><span class="p">);</span>
        <span class="n">System</span><span class="p">.</span><span class="na">setProperty</span><span class="p">(</span><span class="s">"java.security.auth.login.config"</span><span class="p">,</span> <span class="s">"conf/kafka_server_jaas.conf"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"security.protocol"</span><span class="p">,</span> <span class="s">"SASL_PLAINTEXT"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"sasl.mechanism"</span><span class="p">,</span> <span class="s">"PLAIN"</span><span class="p">);</span>

        <span class="n">KafkaConsumer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="n">kafkaConsumer</span> <span class="o">=</span> <span class="k">new</span> <span class="n">KafkaConsumer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span><span class="p">(</span><span class="n">properties</span><span class="p">);</span>
        <span class="n">kafkaConsumer</span><span class="p">.</span><span class="na">subscribe</span><span class="p">(</span><span class="n">Arrays</span><span class="p">.</span><span class="na">asList</span><span class="p">(</span><span class="n">topic</span><span class="p">));</span>
        <span class="k">return</span> <span class="n">kafkaConsumer</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="cm">/***</span>
<span class="cm">     * 生成生产者并返回</span>
<span class="cm">     * @return</span>
<span class="cm">     */</span>
    <span class="kd">public</span> <span class="kd">static</span> <span class="n">KafkaProducer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span> <span class="nf">createProducer</span><span class="p">()</span> <span class="p">{</span>
        <span class="n">logger</span><span class="p">.</span><span class="na">info</span><span class="p">(</span><span class="s">"Create kafka producer"</span><span class="p">);</span>
        <span class="n">Properties</span> <span class="n">properties</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="p">();</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"bootstrap.servers"</span><span class="p">,</span> <span class="n">Holder</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="s">"kafka.host"</span><span class="p">));</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"acks"</span><span class="p">,</span> <span class="s">"all"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"retries"</span><span class="p">,</span> <span class="mi">0</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"batch.size"</span><span class="p">,</span> <span class="mi">16384</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"linger.ms"</span><span class="p">,</span> <span class="mi">1</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"buffer.memory"</span><span class="p">,</span> <span class="mi">33554432</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"key.serializer"</span><span class="p">,</span> <span class="s">"org.apache.kafka.common.serialization.StringSerializer"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"value.serializer"</span><span class="p">,</span> <span class="s">"org.apache.kafka.common.serialization.StringSerializer"</span><span class="p">);</span>
        <span class="n">System</span><span class="p">.</span><span class="na">setProperty</span><span class="p">(</span><span class="s">"java.security.auth.login.config"</span><span class="p">,</span> <span class="s">"conf/kafka_client_jaas.conf"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"security.protocol"</span><span class="p">,</span> <span class="s">"SASL_PLAINTEXT"</span><span class="p">);</span>
        <span class="n">properties</span><span class="p">.</span><span class="na">put</span><span class="p">(</span><span class="s">"sasl.mechanism"</span><span class="p">,</span> <span class="s">"PLAIN"</span><span class="p">);</span>
        <span class="k">return</span> <span class="k">new</span> <span class="n">KafkaProducer</span><span class="o">&lt;</span><span class="n">String</span><span class="p">,</span> <span class="n">String</span><span class="o">&gt;</span><span class="p">(</span><span class="n">properties</span><span class="p">);</span>
    <span class="p">}</span>
</code></pre>
     </div>
     <h2 id="_10">
      参考链接
     </h2>
     <ul>
      <li>
       <a href="https://developer.ibm.com/tutorials/kafka-authn-authz/">
        User authentication and authorization in Apache Kafka
       </a>
      </li>
      <li>
       <a href="https://blog.csdn.net/Sjdjjdnfjf/article/details/78386726">
        Java Api Consumer 连接启用Kerberos认证的Kafka
       </a>
      </li>
      <li>
       <a href="https://blog.csdn.net/tototuzuoquan/article/details/73430874">
        Kafka集群安装配置，kafka后台运行的方式，Kafka配置文件中的参数说明
       </a>
      </li>
      <li>
       <a href="https://kafka.apache.org/documentation/#quickstart">
        Documentation
       </a>
      </li>
      <li>
       <a href="https://blog.csdn.net/u010416101/article/details/81165578">
        Kafka常用命令(带SASL权限版)
       </a>
      </li>
      <li>
       <a href="https://www.jianshu.com/p/102606f8a795">
        Kafka 0.10.0 SASL/PLAIN身份认证及权限实现
       </a>
      </li>
      <li>
       <a href="http://www.voidcn.com/article/p-vkewulnr-bqg.html">
        Kakfa-SASL身份验证登陆
       </a>
      </li>
     </ul>
    </article>
   </div>
   <!-- Footer -->
   <footer id="footer">
    <p class="copyright">
     © Untitled. Design:
     <a href="https://html5up.net">
      HTML5 UP
     </a>
     .
    </p>
   </footer>
  </div>
  <!-- BG -->
  <div id="bg">
  </div>
  <!-- Scripts -->
  <script src="../assets/js/jquery.min.js">
  </script>
  <script src="../assets/js/browser.min.js">
  </script>
  <script src="../assets/js/breakpoints.min.js">
  </script>
  <script src="../assets/js/util.js">
  </script>
  <script src="../assets/js/main.js">
  </script>
 </body>
</html>
